package com.atguigu.edu.realtime.util;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;

/**
 * ClassName: MyKafkaUtil
 * Package: com.atguigu.edu.realtime.util
 * Description:
 * 操作kafka的工具类
 * @Author Mr.2
 * @Create 2023/9/7 16:45
 * @Version 1.0
 */
public class MyKafkaUtil {
    // Kafka 集群地址
    private static final String KAFKA_SERVER = "localhost:9092,localhost:9093,localhost:9094";

    // 1. 获取 kafkaSource
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setTopics(topic)
                .setGroupId(groupId)
                .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")
                // 消费最新位置
                .setStartingOffsets(OffsetsInitializer.latest())
                // 处理 kafka空消息 new DeserializationSchema<String>()
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();

        return source;
    }

    // 2. 获取 kafkaSink
    public static KafkaSink<String> getKafkaSink(String topic) {

        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                // 重要知识点:
                //  在生产环境中，如果要想保证写入的一致性，需要进行如下设置:
                // 1.指定DeliveryGuarantee.EXACTLY_ONCE
                // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 2. 设置 事务前缀 --setTransactionalIdPrefix("")
                // .setTransactionalIdPrefix("")
                // 3. 检查点开启 -- env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
                // 4. 事务超时时间大于检查点超时时间 ， 但是要小于事务最大超时时间(默认15min) -->
                //    扩展: 目前涉及到2个kafka的常量类，producerConfig、consumerConfig,
                //      等同于 kafka官网的 CONFIGURATION -- PRODUCER CONFIGS、CONSUMER CONFIG 里面的一些配置, 如：transaction.timeout.ms
                //.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15*60*1000 + "")
                // 5. 在消费端，对于没有提交的消息，不应该被消费 -- 设置，事务隔离级别 读已提交 'read_committed'
                //.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        return sink;
    }

    // 3. SQL 方式 kafka-connector
    public static String getTopicDbDDL(String groupId) {

        return " CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `ts` STRING,\n" +
                "  `data` MAP<STRING, STRING>,\n" +
                "  `old` MAP<STRING, STRING>,\n" +
                "  proc_time AS PROCTIME()\n" +
                " ) " + getKafkaDDL("topic_db", groupId);
    }

    // 4. SQL 方式 kafka-connector 连接选项
    public static String getKafkaDDL(String topic, String groupId) {
        return "WITH (\n" +
                "   'connector' = 'kafka',\n" +
                "   'topic' = '" + topic + "',\n" +
                "   'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "   'properties.group.id' = '" + groupId + "',\n" +
                "   'scan.startup.mode' = 'latest-offset',\n" +
                "   'format' = 'json'\n" +
                " ) ";
    }

    // 5. SQL 方式 upsert kafka-connector
    public static String getUpsertKafkaDDL(String topic) {
        return " WITH (\n" +
                "   'connector' = 'upsert-kafka',\n" +
                "   'topic' = '" + topic + "',\n" +
                "   'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "   'key.format' = 'json',\n" +
                "   'value.format' = 'json'\n" +
                " )";
    }

}
